//package com.dc.mqtt.newMqtt;
//
//import com.alibaba.fastjson.JSON;
//import com.dc.controller.Controller;
//import com.dc.entity.MessageLog;
//import com.dc.util.ApplicationContextProvider;
//import com.dc.util.SnowflakeIdWorker;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.commons.beanutils.ConvertUtils;
//import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
//import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
//import org.eclipse.paho.client.mqttv3.MqttException;
//import org.eclipse.paho.client.mqttv3.MqttMessage;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.stereotype.Component;
//
//import java.util.List;
//import java.util.Map;
//
//@Slf4j
//public class LogMQTTCallback implements MqttCallbackExtended {
//
//
//    private static final Logger LOGGER = LoggerFactory.getLogger(LogMQTTCallback.class);
//
//    private MyMQTTClient myMQTTClient;
//
//    public LogMQTTCallback(MyMQTTClient myMQTTClient) {
//        this.myMQTTClient = myMQTTClient;
//    }
//
//    /**
//     * 丢失连接，可在这里做重连
//     * 只会调用一次
//     *
//     * @param throwable
//     */
//    @Override
//    public void connectionLost(Throwable throwable) {
//        LOGGER.error("连接断开，下面做重连...");
//        long reconnectTimes = 1;
//        while (true) {
//            try {
//                if (MyMQTTClient.getClient().isConnected()) {
//                    LOGGER.warn("mqtt reconnect success end");
//                    return;
//                }
//                LOGGER.warn("mqtt reconnect times = {} try again...", reconnectTimes++);
//                MyMQTTClient.getClient().reconnect();
//            } catch (MqttException e) {
//                LOGGER.error("", e);
//            }
//            try {
//                MyMQTTClient.getClient().wait(1000);
//            } catch (InterruptedException e1) {
//                //e1.printStackTrace();
//            }
//
//        }
//    }
//
//    /**
//     * 消息到达后
//     * subscribe后，执行的回调函数
//     *
//     * @param s
//     * @param mqttMessage
//     * @throws Exception
//     */
//    @Override
//    public void messageArrived(String s, MqttMessage mqttMessage) {
//        Controller controller = ApplicationContextProvider.getBean(Controller.class);
//        // subscribe后得到的消息会执行到这里面
//        LOGGER.info("接收消息主题 : {}，接收消息内容 : {}", s, new String(mqttMessage.getPayload()));
//        String content = new String(mqttMessage.getPayload());
//        try {
//            MessageLog messageLog = new MessageLog();
//            //TODO 根据设备id写入数据库
//            Map<String, Object> commonParams = JSON.parseObject(content, Map.class);
//            String id = String.valueOf(commonParams.get("id"));
////            if (redisUtil.hasKey(id)) {
////                //如果key存在 就不做任何操作,抛弃当前数据
////            } else {
////                redisUtil.set(id, id, 2L);
//            //不存在，就代表是新数据，保存入库
//            List<MessageLog> data = controller.queryByBu(id);
////                if (CollectionUtils.isEmpty(data)) {
//            messageLog.setBatchId(String.valueOf(commonParams.get("id")));
//            messageLog.setDeviceId(String.valueOf(SnowflakeIdWorker.generateId()));
//            messageLog.setData(content);
//            messageLog.setCountData("0");
//            controller.save(messageLog);
////                }
////            }
//        } catch (Exception exception) {
//            //有可能serviceId为空 有可能serviceId不对拿不到handler
//            log.error("处理失败", exception);
//        }
//    }
//
//    /**
//     * publish后，配送完成后回调的方法
//     *
//     * @param iMqttDeliveryToken
//     */
//    @Override
//    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
//        LOGGER.info("==========deliveryComplete= {} ==========", iMqttDeliveryToken.isComplete());
//    }
//
//    /**
//     * 重连之后订阅主题
//     */
//    @Override
//    public void connectComplete(boolean b, String s) {
//        try {
//            MqttConfiguration mqttConfiguration = ApplicationContextProvider.getBean(MqttConfiguration.class);
//            int[] convert = (int[]) ConvertUtils.convert(mqttConfiguration.qos.split(","), int.class);
//            MyMQTTClient.getClient().subscribe(mqttConfiguration.SUB_TOPICS.split(","), convert); //重连后订阅需要的主题
//        } catch (MqttException e) {
//            e.printStackTrace();
//        }
//    }
//}
